package src;
import java.util.HashSet;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;

public class TinyThreadPool {
    // 1.维护内部的线程引用
    private final HashSet<PullWorker> workers = new HashSet<PullWorker>();
    // 2.任务队列
    private final BlockingQueue<Runnable> taskQueue;
    // 3.线程数相关配置，本线程池实现默认热启动corePoolSize的线程数
    private final int corePoolSize;
    private final int maxPoolSize;
    private final int maxQueueSize;

    /**
     * 4.线程池状态
     * 0 初始化
     * 1 可运行
     * 2 运行中
     * 3 关闭
     */
    static enum PoolState {
        Initial,
        Runnable,
        // Running,
        Shutdown,
    }

    private volatile PoolState poolState = PoolState.Initial; // 并发情况下修改为volatile，多个线程会读取这个变量
    // 5.工厂模式，用于自定义产生线程
    private final ThreadFactory threadFactory;

    public TinyThreadPool(int corePoolSize, int maxPoolSize, int maxQueueSize, ThreadFactory threadFactory,
            BlockingQueue<Runnable> taskQueue) {
        if (corePoolSize < 0) {
            throw new IllegalArgumentException("线程池核心线程数不能小于0！");
        }
        if (corePoolSize > maxPoolSize) {
            throw new IllegalArgumentException("线程池最大线程数不能小于核心线程数！");
        }
        this.corePoolSize = corePoolSize;
        this.maxPoolSize = maxPoolSize;
        this.maxQueueSize = maxQueueSize;
        this.threadFactory = threadFactory;
        this.taskQueue = taskQueue;
        synchronized (workers) {
            for (int i = 0; i < this.corePoolSize; i++) {
                // 热启动，装载corePoolSize的线程数
                PullWorker curWorker = new PullWorker(null);
                workers.add(curWorker);
                curWorker.thread.start();
            }
        }
        this.poolState = PoolState.Runnable;
    }

    public TinyThreadPool(int corePoolSize, int maxPoolSize, int maxQueueSize) {
        // 注意BlockingQueue与BlockingDeque的区别，后者是双端操作
        this(corePoolSize, maxPoolSize, maxQueueSize, new DefaultThreadFactory(),
                new LinkedBlockingQueue<Runnable>(maxQueueSize));
    }

    public void execute(Runnable newTask) {
        System.out.printf("线程池提交任务:%s\n", newTask);
        if (newTask == null) {
            throw new NullPointerException("提交至线程池的任务不能为null！");
        }
        synchronized (this.poolState) {
            if (this.poolState == PoolState.Shutdown) {
                throw new RejectedExecutionException("线程池已关闭，无法再提交任务！");
            }
        }
        // 由于线程池是热启动，所以只需往任务队列中添加任务即可
        if (!taskQueue.offer(newTask)) {
            synchronized (workers) {
                if (workers.size() < maxPoolSize) {
                    PullWorker curWorker = new PullWorker(newTask);
                    workers.add(curWorker);
                    curWorker.thread.start();
                } else {
                    throw new RejectedExecutionException(String.format("线程池已达maxPoolSize:%d,并且任务队列已满！", maxPoolSize));
                }
            }
        }
    }

    public void shutdown() {
        synchronized (this.poolState) {
            poolState = PoolState.Shutdown;
        }
        synchronized (workers) {
            for (PullWorker curWorker : workers) {
                curWorker.thread.interrupt();
            }
        }
    }
    //获取目前线程池内部的线程数量
    public int getThreadSize() {
        synchronized (workers) {
            return workers.size();
        }
    }
    //获取线程池内部队列大小
    public int getQueueSize() {
        return taskQueue.size();
    }

    /**
     * PullWorker，实际的状态机，可以执行代码逻辑的单元
     */
    private class PullWorker implements Runnable {
        final Thread thread;
        Runnable curTask;

        public PullWorker(Runnable task) {
            this.curTask = task;
            this.thread = threadFactory.newThread(this); // 之前错误传为this
        }

        @Override
        public void run() {
            try {
                Runnable taskToDeal = curTask;
                curTask = null;
                System.out.printf("工作线程:%s开始运行\n",this);
                /**
                 * 1.首先运行本地的任务
                 * 2.从队列里面获取任务，开始执行
                 * 线程池关闭时，内部的PullWorker线程会收到中断，停止循环
                 */
                while (taskToDeal != null || (taskToDeal = getTaskFromQueue()) != null) {
                    // 由于此处是业务逻辑，可能抛出异常，这里用try-catch进行包裹，防止业务逻辑代码影响工作线程消费任务
                    System.out.printf("工作线程:%s获取任务成功%s\n",this,taskToDeal);
                    try {
                        taskToDeal.run();
                    } catch (Exception e) {
                        System.err.printf("PullWorker内层异常，当前Worker信息：%s，异常信息:%s", this, e.getMessage());
                        e.printStackTrace();
                    } finally {
                        taskToDeal = null;
                    }
                }
            } catch (Exception e) {
                System.err.printf("PullWorker外层异常，当前Worker信息：%s，异常信息:%s", this, e.getMessage());
                e.printStackTrace();
            } finally {
                /**
                 * 由于我们在内部已经处理了Runnable可能抛出的异常，这里的异常一般是InterruptedException
                 * 由外部的线程池调用，用来终止内层的while循环，停掉工作线程
                 */
                synchronized (workers) {
                    workers.remove(this);
                }
            }

        }

        private Runnable getTaskFromQueue() throws IllegalStateException {
            try {
                // synchronized (poolState) {
                // WARNING 这里poolState的可见性由volatile读取，不能用同步块，否则会出现死锁的情况
                // 线程池状态为Shutdown状态，通过poll消费存量的任务，非阻塞
                if (poolState == PoolState.Shutdown) {
                    return taskQueue.poll();
                } else if (poolState == PoolState.Runnable) {
                    // 线程池状态为Runnable，阻塞式消费任务
                    return taskQueue.take();
                } else {
                    throw new IllegalStateException(String.format("线程池状态扭转异常:%s", poolState));
                }
            } catch (InterruptedException e) {
                return null;
            }
        }
    }

    static class DefaultThreadFactory implements ThreadFactory {
        private static int curThreadCount = 0;

        @Override
        public Thread newThread(Runnable r) {
            curThreadCount++;
            return new Thread(r, String.format("TinyThreadPool PullWorker-%d", curThreadCount));
        }
    }
}